package cn.edu.flink.scala.tutorial.window

import cn.edu.flink.scala.tutorial.source.Event
import com.sun.xml.internal.ws.model.RuntimeModeler
import org.apache.flink.api.common.RuntimeExecutionMode
import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object AllowedLatenessTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)

    // 本地变量无效果,需要控制数据的时间间隔
    val events = List(
      Event("Mary", "home/", 1000L),
      Event("Mary", "cart/", 2000L),
      Event("Mary", "cart/", 3000L),
      Event("Mary", "cart/", 10000L),
      Event("Mary", "cart/", 6000L),
      Event("Mary", "cart/", 20000L),
      Event("Mary", "cart/", 30000L),
      Event("Mary", "cart/", 40000L))

    val eventsDS = env.addSource(new EventsSource(events))
      .assignAscendingTimestamps(_.timestamp)

    eventsDS.keyBy(_.user)
      .window(TumblingEventTimeWindows.of(Time.seconds(10)))
      // 指定窗口等待时间
      .allowedLateness(Time.seconds(5))
      .process(new WatermarkWindowResult)
      .print()

    env.execute("AllowedLatenessTest")
  }

  class EventsSource(var events: List[Event]) extends SourceFunction[Event] {
    override def run(ctx: SourceFunction.SourceContext[Event]): Unit = {
      for (elem <- events) {
        ctx.collect(elem)
        Thread.sleep(1000)
      }
    }

    override def cancel(): Unit = ???
  }

  class WatermarkWindowResult extends ProcessWindowFunction[Event, String, String, TimeWindow] {
    override def process(user: String, context: Context, elements: Iterable[Event], out: Collector[String]): Unit = {
      // 提取信息
      val start = context.window.getStart
      val end = context.window.getEnd
      val count = elements.size

      // 增加水位线信息
      val currentWatermark = context.currentWatermark

      out.collect(s"窗口 $start ~ $end , 用户 $user 的活跃度为：$count, 水位线现在位于：$currentWatermark")
    }
  }
}
